package cn.itcast.batch;

import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.typeutils.RowTypeInfo;

import java.io.Serializable;
import java.util.Objects;

//flinkJDBC操作抽象类
public class JDBCFormatAbstract implements Serializable {
    /**
     * 批的方式实现数据读取
     * @param driverName
     * @param url
     * @param userName
     * @param password
     * @param inputSql  sql语句
     * @param rowTypeInfo  表结构
     * @return
     */
    public JDBCInputFormat getBatchJDBCInputFormat(String driverName, String url, String userName, String password, String inputSql, RowTypeInfo rowTypeInfo){
        JDBCInputFormat jdbcInputFormat = null;
        JDBCInputFormat.JDBCInputFormatBuilder inputFormatBuilder = JDBCInputFormat.buildJDBCInputFormat()
                .setDrivername(driverName)
                .setDBUrl(url)
                .setUsername(userName)
                .setPassword(password)
                .setQuery(inputSql)
                .setRowTypeInfo(rowTypeInfo);

        jdbcInputFormat = inputFormatBuilder.finish();

        return jdbcInputFormat;
    }


    /**
     * 批的方式写入数据
     * @param driverName
     * @param url
     * @param userName
     * @param password
     * @param outputSql
     * @param sqlType
     * @return
     */
    public JDBCOutputFormat getBatchJDBCOutputFormat(String driverName, String url, String userName, String password, String outputSql, int[] sqlType){
        JDBCOutputFormat jdbcOutputFormat = null;
        JDBCOutputFormat.JDBCOutputFormatBuilder outputFormatBuilder = JDBCOutputFormat.buildJDBCOutputFormat()
                .setDrivername(driverName)
                .setDBUrl(url)
                .setUsername(userName)
                .setPassword(password)
                .setQuery(outputSql);

        if(Objects.nonNull(sqlType)){//如果不为空追加参数
            outputFormatBuilder.setSqlTypes(sqlType);
            jdbcOutputFormat = outputFormatBuilder.finish();
        }else {
            jdbcOutputFormat = outputFormatBuilder.finish();
        }

        return jdbcOutputFormat;
    }
}
